/*=============================================================================
#     FileName: srv_main.cpp
#         Desc: 主函数
#       Author: Ace
#        Email: goace@acm.org
#     HomePage: http://www.goace.co.cc
#      Version: 0.0.1
#   LastChange: 2011-05-30 18:09:57
#      History:
=============================================================================*/
#pragma		pack(4)

#ifndef		_GNU_SOURCE
#define     _GNU_SOURCE
#endif

#include  <sys/mman.h>
#include  <sys/epoll.h>
#include  <stdlib.h>
#include  <unistd.h>
#include  <fcntl.h>
#include  <errno.h>
#include  <sys/stat.h>
#include  <sys/types.h>
#include  <sys/wait.h>
#include  <stdio.h>
#include  <string.h>
#include  <iostream>
#include  <semaphore.h>
#include  <sys/sendfile.h>
#include  <sys/socket.h>
#include  <netinet/in.h>
#include  <netinet/tcp.h>
#include  "db.h"
#include  "my_tcp_lib.h"
#include  "gserver.h"
#include  "event_handler.h"
#include  "my_sock_lib.h"
#include  <map>
using namespace std;
using namespace ace;

//全局变量
DB								db;						//数据库对象
sem_t							ctl_have_events;		//信号量，同步ctl进程中的线程读取事件
int								n_ctl_idle_threads;		//记录空闲的ctl线程
map<string,ctl_sock_info_t*>	online;					//记录用户在线情况，实现单点登录
map<string,ctl_sock_info_t*>	*online_info;


/* ---------------------------------------------------------------*/
/**
 * 函数简介:  建立工作进程，master永不返回，
 *            始终监视子进程的情况，如果子
 *            进程意外crash，就重新建立。
 *
 * 参数: nWorker 要建立的进程数
 *
 * 返回值:  -1表示错误   0表示成功
 *          master进程不会返回
 */
/* ---------------------------------------------------------------*/
int create_workers(int nWorker, int flag)
{
    if (0 < nWorker)
    {
        bool bIsChild;
        pid_t nPid;

        while (!bIsChild)
        {
            if (0 < nWorker)
            {
                nPid = ::fork();
                if (nPid > 0)
                {
                    bIsChild = false;
                    --nWorker;
                }
                else if (0 == nPid)
                {
                    bIsChild = true;
					const char * type = flag == 1 ? "ctl" : "data";
                    printf("create %s worker %d success!\n", type, getpid());
                }
                else
                {
                    printf("fork error: %s\n", strerror(errno));
                    return -1;
                }
            }
            else 
            {
                int nStatus;
                if (-1 == wait(&nStatus))
                {
                    ++nWorker;
                }
            }
        }
    }

    return 0;
}

/* ---------------------------------------------------------------*/
/**
 * 函数简介:  初始化进程池
 *
 * 参数： n_child 要创建的workers数
 */
/* ---------------------------------------------------------------*/
void data_workers(int n_child)
{
	void data_process(int);

	int listen_sock;
	listen_sock = tcp_listen(NULL,SRV_DATA_PORT,NULL);	
	if( listen_sock < 0 )
	{
		perror("tcp_listen");
		master_exit(-1);
	}

	setnonblocking(listen_sock);
	
	//创建工作进程
	create_workers(n_child,2);

	data_process(listen_sock);
}

/* ---------------------------------------------------------------*/
/**
 * 函数简介: 数据处理进程的主例程 
 *
 * 三种请求类型的状态机如下：
 *   SYNC： SYNC_WAIT_ID  ->  SYNC_WAIT_VERSION  -> SYNC_SENDING
 *   PUSH:  PUSH_WAIT_ID  ->  PUSH_WAIT_ACK  ->  PUSH_SENDING
 *              ->  PUSH_RECIVING
 *   PULL:  PULL_WAIT_ID  ->  PULL_RECIVING  ->  PULL_WAIT_ACK
 *              ->  PULL_SENDING
 *
 * 参数: listen_sock 监听的socket
 */
/* ---------------------------------------------------------------*/
void data_process(int listen_sock)
{
	int8_t		reply;					//给用户的反馈
	char		name[USR_LEN], pwd[PWD_LEN];
	uint8_t		request_type;
	int32_t		res;

	struct epoll_event ev,events[EPOLL_MAX_EVENTS];
	data_sock_info_t *ptr;

	int nfds,							//计数epoll返回的事件数目
		connfd,							//accept返回的fd
		n_read;							//read返回的数	
	int epfd = epoll_create(EPOLL_MAX_FD);
	if( epfd < 0 )
	{
		data_exit(-1);
	}

	//把监听socket加入epoll集合中
	ev.events = EPOLLIN | EPOLLET;
	ev.data.ptr = new data_sock_info_t;
	((data_sock_info_t*)ev.data.ptr)->sock_fd = listen_sock;

	if( epoll_ctl(epfd,EPOLL_CTL_ADD,listen_sock,&ev) < 0 )
	{
		perror("data_process epoll_ctl add listen sock");
		data_exit(-1);
	}

	//main loop
	while( true )
	{
		nfds = epoll_wait(epfd,events,EPOLL_MAX_EVENTS,-1);
		if( nfds == -1 )
		{
			if( errno == EINTR )
				continue;
			else
			{
				perror("data_process epoll_wait");
				ctl_exit(-1);
			}
		}

		//循环处理每个返回的event
		for( int i = 0 ; i < nfds ; i++ )
		{
			ptr = (data_sock_info_t*)events[i].data.ptr;

			if( ptr->sock_fd == listen_sock )
			{
				//因为设置的ET模式，因此要循环accept直到EAGAIN
				while( true )
				{
					connfd = accept(listen_sock,NULL,NULL);
					if( connfd == -1 )
					{
						if( errno == EINTR )
							continue;
						else if( errno == EAGAIN || errno == EWOULDBLOCK)
							break;
						else if( errno == ECONNABORTED)
							break;
						else
						{
							perror("data_process accept");
							continue;
						}
					}
					
					//把返回的socket加入epoll集合
					set_sock_timeo(connfd,READ|WRITE);
					setkeepalive(connfd);
					setnonblocking(connfd);
					ev.events = EPOLLIN | EPOLLET |  EPOLLRDHUP;

					//分配结构体
					ev.data.ptr = new data_sock_info_t;
					bzero(ev.data.ptr,sizeof(data_sock_info_t));
					((data_sock_info_t*)ev.data.ptr)->sock_fd = connfd;
					((data_sock_info_t*)ev.data.ptr)->type = DATA_INIT;
					
					if( epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev) < 0 )
					{
						perror("data_process after accept epoll_ctl");
						close(connfd);
						delete (data_sock_info_t*)ev.data.ptr;
						continue;
					}
				}
			}
			else	//如果是非监听socket的事件
			{
				if( events[i].events & EPOLLIN )	//如果是可读
				{
					if( ptr->type == DATA_INIT )	//如果处在INIT状态
					{
						//读取请求类型：SYNC PULL PUSH之一
						if( (res = read(ptr->sock_fd,&ptr->type,sizeof(ptr->type))) <= 0 )
						{
							if(res < 0)
								data_error("data_process read option type from peer",ptr);
							else
							{
								close(ptr->sock_fd);
								delete ptr;
							}


							continue;
						}

						//因为epoll设置了ET模式  所以要继续读  直到EAGAIN
						if( ptr->type == DATA_SYNC )
						{
							//读取用户ID
							if( data_read_id(ptr) < 0 )
								continue;

							//读取客户端版本号
							if( data_read_version(SYNC_WAIT_PEER,ptr) < 0 )
								continue;

							//读取服务器版本号
							if( data_read_version(SYNC_WAIT_REAL,ptr) < 0 )
								continue;
							
							//判定客户端的版本号是否小于服务器
							if( ptr->peer_version > ptr->real_version )
							{
								cout << "user's version is higher than server: " 
										<< ptr->peer_version << " > " << ptr->real_version 
										<<  " no need to sync" << endl;

								close(ptr->sock_fd);
								delete ptr;
								continue;
							}

#ifdef DEBUG
							cout << ptr->usr_name << "\trequest to sync from version " 
									<< ptr->peer_version << endl;
#endif

							ptr->state = SYNC_WAIT_REAL;
							
							//打开日志文件
							if( open_log_file(ptr) < 0 )
								continue;

							ptr->state = SYNC_SENDING;
							ptr->transfering = true;
							ptr->tsfd = 0;

							//等待该socket可写
							ev.events = EPOLLOUT | EPOLLRDHUP | EPOLLET;
							ev.data.ptr = ptr;
							if( epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->sock_fd,&ev) < 0 )
							{
								data_error("data_process during DATA_SYNC epoll_ctl",ptr);
								continue;
							}

						}
						else if( ptr->type == DATA_PUSH )
						{
							//读取用户ID
							if( data_read_id(ptr) < 0 )
								continue;

							//读取文件名 打开要更新的文件
							if( data_read_path(ptr,DATA_PUSH) < 0 )
								continue;
							
							//接收文件
							ssize_t res;
							while((res = splice(ptr->sock_fd,NULL,ptr->p[1],
											NULL,BLOCK_SIZE,SPLICE_F_MOVE|SPLICE_F_NONBLOCK)) > 0 )
							{
								splice(ptr->p[0],NULL,ptr->file_fd,NULL,
										BLOCK_SIZE,SPLICE_F_MOVE|SPLICE_F_NONBLOCK);
							}

							if(res == 0)
							{
								close(ptr->p[0]);
								close(ptr->p[1]);
								close(ptr->sock_fd);
								close(ptr->file_fd);
								delete ptr;
								continue;
							}
							else if(res == -1)
							{
								if( errno == EAGAIN || errno == EWOULDBLOCK )
								{
									ptr->transfering = true;
									ptr->state = RECEIVING;
									continue;
								}
								cout << "res = " << res << "spliceerror1" <<  endl;
								data_error("data_process during DATA_PUSH splice",ptr);
								continue;
							}
						}	 
						else if( ptr->type == DATA_PULL)
						{
							//读取用户ID
							if( data_read_id(ptr) < 0 )
								continue;

							//读取文件名 打开要更新的文件
							if( data_read_path(ptr,DATA_PULL) < 0 )
								continue;

							//发送文件
							ssize_t sent;
							while(true)
							{
								sent = sendfile(ptr->sock_fd,ptr->file_fd,NULL,BLOCK_SIZE);
								if( sent > 0 )  //发送成功
								{
									 ptr->tsfd += sent;
									 if( ptr->tsfd == ptr->total )
									 {
										 sent = 0;
										 break;
									 }
								} 
								else			//发送失败
								{
									break;	 
								}

							}

							if(sent == 0)
							{
								close(ptr->sock_fd);
								close(ptr->file_fd);
								delete ptr;
								continue;
							}
							else if(sent == -1)
							{
								if( errno == EAGAIN || errno == EWOULDBLOCK )
								{
									ev.data.ptr = ptr;
									ev.events = EPOLLOUT|EPOLLET|EPOLLRDHUP;
									if( epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->sock_fd,&ev) < 0 )
									{
										data_error("data_process during DATA_PULL epoll_ctl",ptr);
										continue;
									}
									ptr->transfering = true;
									ptr->state = SENDING;
									continue;
								}
								data_error("data_process during DATA_PULLsendfile",ptr);
								continue;
							}

						}
					}
					else if( ptr->type == DATA_SYNC)
					{
						if( ptr->state == SYNC_WAIT_ID )
						 {
								 //读取用户ID
								 if( data_read_id(ptr) < 0 )
										 continue;

								 ptr->state = SYNC_WAIT_PEER;
						 }
						 
						 if( ptr->state == SYNC_WAIT_PEER )
						 {
								 //读取客户端版本号
								 if( data_read_version(SYNC_WAIT_PEER,ptr) < 0 )
										 continue;
								 ptr->state = SYNC_WAIT_REAL; 
						 }

						 if( ptr->state == SYNC_WAIT_REAL )
						 {
								 //读取服务器版本号
								 if( data_read_version(SYNC_WAIT_REAL,ptr) < 0 )
										 continue;
						 }

						//判定客户端的版本号是否小于服务器
						if( ptr->peer_version > ptr->real_version )
						{
							cout << "user's version is higher than server: " 
									<< ptr->peer_version << " > " << ptr->real_version 
									<<  "no need to sync" << endl;

							close(ptr->sock_fd);
							delete ptr;
							continue;
						}

#ifdef DEBUG
						cout << ptr->usr_name << "\trequest to sync from version " 
								<< ptr->peer_version << endl;
#endif

						//打开日志文件
						 if( open_log_file(ptr) < 0 )
								 continue;

						 ptr->state = SYNC_SENDING;
						 ptr->transfering = true;

						 //等待该socket可写
						 ev.events = EPOLLOUT | EPOLLRDHUP | EPOLLET;
						 ev.data.ptr = ptr;
						 if( epoll_ctl(epfd,EPOLL_CTL_MOD,
												 ptr->sock_fd,&ev) < 0 )
						 {
								 data_error("data_process during DATA_SYNC epoll_ctl",ptr);
								 continue;
						 }
					}
					else if( ptr->type == DATA_PUSH )
					{
						if( ptr->state == WAIT_ID )
						{
							if( data_read_id(ptr) < 0 )
								continue;
						}	

						if( ptr->state == WAIT_PATH )
						{
							if( data_read_path(ptr,DATA_PUSH) < 0 )
								continue;
						}

						if( ptr->state == RECEIVING )
						{
							int res;

							while((res = splice(ptr->sock_fd,NULL,ptr->p[1],
											NULL,BLOCK_SIZE,SPLICE_F_MOVE|SPLICE_F_NONBLOCK)) > 0 )
							{
								splice(ptr->p[0],NULL,ptr->file_fd,NULL,
										BLOCK_SIZE,SPLICE_F_MOVE|SPLICE_F_NONBLOCK);
							}

							if(res == 0)
							{
								close(ptr->p[0]);
								close(ptr->p[1]);
								close(ptr->sock_fd);
								close(ptr->file_fd);
								delete ptr;
								continue;
							}
							else if(res == -1)
							{
								if( errno == EAGAIN || errno == EWOULDBLOCK )
								{
									ptr->transfering = true;
									ptr->state = RECEIVING;
									continue;
								}
								cout << "res = " << res <<" spliceerror2" <<  endl;
								data_error("data_process during DATA_PUSH splice",ptr);
								continue;
							}
						
							continue;
						}
					}
					else if( ptr->type == DATA_PULL )
					{

						if( ptr->state == WAIT_ID )
						{
							if( data_read_id(ptr) < 0 )
								continue;
						}	

						if( ptr->state == WAIT_PATH )
						{
							if( data_read_path(ptr,DATA_PULL) < 0 )
							{
								continue;
							}
						}

						if( ptr->state == SENDING )
						{
							int sent;
							while(true)
							{
								sent = sendfile(ptr->sock_fd,ptr->file_fd,NULL,BLOCK_SIZE);
								if( sent > 0 )  //发送成功
								{
									 ptr->tsfd += sent;
									 if( ptr->tsfd == ptr->total )
									 {
										 sent = 0;
										 break;
									 }
								} 
								else			//发送失败
								{
									break;	 
								}
							}

							if(sent == 0)
							{
								//close(ptr->p[0]);
								//close(ptr->p[1]);
								close(ptr->sock_fd);
								close(ptr->file_fd);
								delete ptr;
								continue;
							}
							else if(sent == -1)
							{
								if( errno == EAGAIN || errno == EWOULDBLOCK )
								{
									ev.data.ptr = ptr;
									ev.events = EPOLLOUT|EPOLLET|EPOLLRDHUP;
									if( epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->sock_fd,&ev) < 0 )
									{
										data_error("data_process during DATA_PULL epoll_ctl",ptr);
										continue;
									}
									ptr->transfering = true;
									ptr->state = SENDING;
									continue;
								}
								data_error("data_process during DATA_PULL sendfile",ptr);
								continue;
							}

						}

					}
				}
				else if( events[i].events & EPOLLOUT )
				{
					if( ptr->type == DATA_SYNC)
					{
						ssize_t sent;

						while( ptr->tsfd < ptr->total )
						{
							if(( sent = sendfile(ptr->sock_fd,ptr->file_fd,
										NULL, BLOCK_SIZE)) < 0 )
							{
								if( errno == EAGAIN || errno == EWOULDBLOCK)
									break;

								data_error("data_process during DATA_SYNC sendfile",ptr);
								break;
							}
							ptr->tsfd += sent;

							if( ptr->tsfd == ptr->total )
							{
								//如果所有log传送完了
								if( ptr->peer_version - 1 == ptr->real_version )
								{
									//cout << "log传送完毕" << endl;
									close(ptr->file_fd);
									close(ptr->sock_fd);
									delete ptr;
									break;
								}
									
								//打开下一个log文件
								close(ptr->file_fd);
								if( open_log_file(ptr) < 0 )
								{
									break;
								}
							}
						}
						
					}
					else if( ptr->type == DATA_PUSH )
					{
						
					}
					else if( ptr->type == DATA_PULL )
					{
						
						if( ptr->state == SENDING )
						{
							ssize_t sent;
							while(true)
							{
								sent = sendfile(ptr->sock_fd,ptr->file_fd,NULL,BLOCK_SIZE);
								if( sent > 0 )  //发送成功
								{
									 ptr->tsfd += sent;
									 if( ptr->tsfd == ptr->total )
									 {
										 sent = 0;
										 break;
									 }
								} 
								else			//发送失败
								{
									break;	 
								}
							}

							if(sent == 0)
							{
								//close(ptr->p[0]);
								//close(ptr->p[1]);
								close(ptr->sock_fd);
								close(ptr->file_fd);
								delete ptr;
								continue;
							}
							else if(sent == -1)
							{
								if( errno == EAGAIN || errno == EWOULDBLOCK )
								{
									continue;
								}
								data_error("data_process during DATA_PULL sendfile",ptr);
								continue;
							}
						}
					}
				}
				else if( events[i].events & EPOLLERR)
				{
					//返回EPOLLERR事件说明
					//客户端心跳超时 
					//于是可以判定用户掉线

					close(ptr->sock_fd);

					if(ptr->transfering)
					{
						close(ptr->file_fd);
						if( ptr->type == DATA_PUSH )
						{
							close(ptr->p[0]);
							close(ptr->p[1]);
						}
					}

					delete ptr;
				}
				else if( events[i].events & EPOLLRDHUP )
				{
					//对方关闭socket
					close(ptr->sock_fd);

					if(ptr->transfering)
					{
						close(ptr->file_fd);
						if( ptr->type == DATA_PUSH )
						{
							close(ptr->p[0]);
							close(ptr->p[1]);
						}
					}

					delete ptr;
				}
				else
				{

				}
			}
		}
	}
}

/* ---------------------------------------------------------------*/
/**
 * 函数简介:  循环接收并处理用户“控制信息”，包括注册、登录、
 *			  删除、新建
 *
 * 参数: listen_sock_ctl	 监听控制连接请求的socket
 *       listen_sock_notify	 监听通知连接请求的socket
 */
/* ---------------------------------------------------------------*/
void ctl_process(int listen_sock_ctl, int listen_sock_notify)
{
	int32_t		reply;									//给用户的反馈
	char		name[USR_LEN], pwd[PWD_LEN];
	
	struct epoll_event ev,events[EPOLL_MAX_EVENTS];
	ctl_sock_info_t *ptr;								//记录用户状态的结构体

	int nfds,											//计数epoll返回的事件数目
		connfd,											//accept返回的fd
		n_read;											//read返回的数	

	
	//struct sockaddr	peer_addr;
	//socklen_t	sock_len = sizeof(peer_addr);

	//创建epoll句柄
	int epfd = epoll_create(EPOLL_MAX_FD);
	if( epfd < 0 )
	{
		perror("ctl_precess -> epoll_create");
		ctl_exit(-1);
	}

	//把监听socket加入epoll集合中
	ev.events = EPOLLIN | EPOLLET;
	ev.data.ptr = new ctl_sock_info_t;
	((ctl_sock_info_t*)ev.data.ptr)->fd = listen_sock_ctl;

	if( epoll_ctl(epfd,EPOLL_CTL_ADD,listen_sock_ctl,&ev) < 0 )
	{
		perror("ctl_process epoll_ctl add listen sock");
		ctl_exit(-1);
	}

	ev.data.ptr = new ctl_sock_info_t;
	((ctl_sock_info_t*)ev.data.ptr)->fd = listen_sock_notify;

	if( epoll_ctl(epfd,EPOLL_CTL_ADD,listen_sock_notify,&ev) < 0 )
	{
		perror("ctl_process epoll_ctl add listen sock");
		ctl_exit(-1);
	}

	//主循环
	while( true )
	{
		nfds = epoll_wait(epfd,events,EPOLL_MAX_EVENTS,-1);
		if( nfds == -1 )
		{
			if( errno == EINTR )
				continue;
			else
			{
				perror("ctl_process epoll_wait");
				ctl_exit(-1);
			}
		}

		//循环处理每个返回的event
		for( int i = 0 ; i < nfds ; i++ )
		{
			ptr = (ctl_sock_info_t *)events[i].data.ptr;

			//如果返回的是监听socket
			if( ptr->fd == listen_sock_ctl )
			{
				//因为设置的ET模式，因此要循环accept直到EAGAIN
				while( true )
				{
				
					connfd = accept(ptr->fd, NULL, NULL);
					if( connfd == -1 )
					{
						if( errno == EINTR )
							continue;
						else if( errno == EAGAIN || errno == EWOULDBLOCK)
							break;
						else if( errno == ECONNABORTED)
							break;
						else
						{
							perror("ctl_process accept");
							continue;
						}
					}

					//设置socket的超时
					if( set_sock_timeo(connfd,READ|WRITE) < 0)
					{
						close(connfd);
						continue;
					}

					//设置心跳
					if(setkeepalive(connfd) < 0)
					{
						close(connfd);
						continue;
					}

					//设置非阻塞
					if(setnonblocking(connfd) < 0)
					{
						close(connfd);
						continue;
					}
					
					ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;

					//分配结构体
					ev.data.ptr = new ctl_sock_info_t;
					bzero(ev.data.ptr,sizeof(ctl_sock_info_t));
					((ctl_sock_info_t*)ev.data.ptr)->fd = connfd;
				//	((ctl_sock_info_t*)ev.data.ptr)->peer_addr = peer_addr;
					
					//把返回的socket加入epoll集合
					if( epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev) < 0 )
					{
						perror("ctl_process -> 把accept返回的socket加入epoll");
						close(connfd);
						continue;
					}
				}
			}
			else if(ptr->fd == listen_sock_notify)
			{
				//因为设置的ET模式，因此要循环accept直到EAGAIN
				while( true )
				{
					connfd = accept(ptr->fd, NULL, NULL);
					if( connfd == -1 )
					{
						if( errno == EINTR )
							continue;
						else if( errno == EAGAIN || errno == EWOULDBLOCK)
							break;
						else if( errno == ECONNABORTED)
							break;
						else
						{
							perror("ctl_process -> accept");
							continue;
						}
					}
					
					//设置socket的超时
					if( set_sock_timeo(connfd,READ|WRITE) < 0)
					{
						close(connfd);
						continue;
					}

					//设置心跳
					if(setkeepalive(connfd) < 0)
					{
						close(connfd);
						continue;
					}

					//接受用户名
					char usr_name[USR_LEN];
					if( readn(connfd,usr_name,USR_LEN) < 0 )
					{
						perror("ctl_process after accept notify_sock read" );
						close(connfd);
						continue;
					}


					//保存notify连接
					map<string,ctl_sock_info_t*>::iterator itr = online_info->find(usr_name);
					if(itr != online_info->end())
					{
						itr->second->notify_fd = connfd;
					}
						//(*online_info)[usr_name]->notify_fd = connfd;
				}

			}
			else
			{
				if( events[i].events & EPOLLIN )
				{
					ctl_pkg_t *pkg_in = new ctl_pkg_t;

					setblocking(ptr->fd);
					if ( (n_read = readn(ptr->fd,pkg_in,sizeof(ctl_pkg_t))) <= 0)
					{
						if( n_read == 0 )
						{
							//返回0说明对方关闭了连接
							usr_exit(ptr);
							continue;
						}
						else
						{
							//perror("ctl_process -> read from client");
							usr_exit(ptr);
							continue;
						}
					}
					setnonblocking(ptr->fd);

					//处理事件
					if( pkg_in->type == REG )
					{
						/*注册账户时
						 *如果写回-1 说明用户名已经存在
						 *如果写回-2 说明发生未知错误
						 */

						//插入数据库
						strncpy(ptr->usr_name,pkg_in->str.id.usr,USR_LEN);
						ptr->reply = db.add_usr(ptr->usr_name,pkg_in->str.id.pwd);
						delete pkg_in;
						
						if(ptr->reply == 0)
						{
#ifdef DEBUG
							cout << ptr->usr_name << "\tregist\tpwd: " 
								 << pkg_in->str.id.pwd << endl;
#endif
							
							//如果注册成功
							//建立用户主目录：/gsync/user_ID
							char path[PATH_LEN];
							snprintf(path,PATH_LEN,"%s/%s",SRV_ROOT,ptr->usr_name);
							
							if( mkdir(path,0766) < 0 )
							{
								cerr << "ctl_process during REG: make usr main_path "
									<< path << endl;
								perror(NULL);
								close(ptr->fd);
								delete ptr;
								continue;
							}

							//创建.log目录，存放log文件
							strncat(path,"/.log",PATH_LEN);
							if( mkdir(path,0766) < 0 )
							{
								cerr << "ctl_process -> during REG: create .log dir  "
									<< path << endl;
								perror(NULL);
								close(ptr->fd);
								delete ptr;
								continue;
							}

							//创建version文件
							snprintf(path,PATH_LEN,"%s/%s/.version",SRV_ROOT,ptr->usr_name);

							int vf_fd = open(path,O_WRONLY|O_CREAT,0766);
							if( vf_fd < 0 )
							{
								cerr << "ctl_process during REG create .version "
									<< path << endl;
								perror(NULL);
								ptr->reply = -2;
							}
							else
							{
								//版本号初始化为0
								int32_t init_version = 0;
								if( write(vf_fd,&init_version,sizeof(int32_t)) < 0 )
								{
									perror("ctl_process duing REG write to version file");
									ptr->reply = -2;
								}
							}
							close(vf_fd);
						}
											
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = REG;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("ctl_process during REG write to user");
						}

						close(ptr->fd);
						delete ptr;
					}
					else if( pkg_in->type == LOGIN ) // 用户登录 
					{
						strncpy(ptr->usr_name,pkg_in->str.id.usr,USR_LEN);


						//查询数据库
						ptr->reply = db.log_in(ptr->usr_name,pkg_in->str.id.pwd);
						delete pkg_in;

						if( ptr->reply == 0)
						{
							//检查用户是否已经登录
							map<string,ctl_sock_info_t*>::iterator it = 
								online_info->find(ptr->usr_name);

							if( it != online_info->end() )
							{
								//说明该用户已经在别的机器登录
								force_offline(it->second);
							}

							//记录新登录的用户信息
							try
							{
								online_info->insert(map<string,ctl_sock_info_t*>::value_type(
										ptr->usr_name, ptr));
							}
							catch(exception &ex)
							{
								cout << ex.what() << endl;
								usr_exit(ptr);
								continue;
							}

							//生成版本号文件的路径
							char version_file[USR_LEN];
							snprintf(version_file,USR_LEN,"%s/%s/.version",
									SRV_ROOT,ptr->usr_name);
							
							//打开并读取版本号
							ptr->updated = false;
							
							int vf_fd = open(version_file,O_RDONLY);
							if( vf_fd < 0 )
							{
								cerr << "ctl_process during LOGIN open version file: " 
										<< version_file;
								perror(NULL);
								usr_exit(ptr);
								continue;
							}

							reply = readn(vf_fd,&(ptr->version),sizeof(int32_t));
							if( reply < 0 )
							{
								perror("ctl_process during LOGIN read from version file");
								close(vf_fd);
								usr_exit(ptr);
								continue;
							}

							close(vf_fd);

#ifdef DEBUG
							cout << ptr->usr_name << "\tlogin from PC at version " 
								 << ptr->version << endl;
#endif

						}

						//发送登录结果给客户
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = LOGIN;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("ctl_process during LOGIN write result to user");
							usr_exit(ptr);
							continue;
						}

						//如果登录失败
						if( ptr->reply < 0 )
						{
							close(ptr->fd);
							delete ptr;
							continue;
						} 
						
						//发送版本号
						setblocking(ptr->fd);
						
						if(write(ptr->fd,&(ptr->version),sizeof(ptr->version)) < 0)
						{
							usr_exit(ptr);
							continue;
						}

						setnonblocking(ptr->fd);
					}
					else if( pkg_in->type == LOGIN_PHONE )		//从手机登录
					{
						//记录用户名
						strncpy(ptr->usr_name,pkg_in->str.id.usr,USR_LEN);

						//查询数据库
						ptr->reply = db.log_in(pkg_in->str.id.usr,
												pkg_in->str.id.pwd);
						delete pkg_in;
						
						if( ptr->reply == 0 )
						{
							//检查用户是否已经登录
							map<string,ctl_sock_info_t*>::iterator it = 
								online_info->find(ptr->usr_name);

							if( it != online_info->end() )
							{
								//说明该用户已经在别的机器登录
								force_offline(it->second);
							}

							//记录新登录的用户信息
							try
							{
								online_info->insert(map<string,ctl_sock_info_t*>::value_type(
										ptr->usr_name, ptr));
							}
							catch(exception &ex)
							{
								cout << ex.what() << endl;
								usr_exit(ptr);
								continue;
							}

#ifdef DEBUG
							cout << ptr->usr_name << "\tlogin from phone "  << endl;
#endif			

						}

						//发送登录结果给客户
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = LOGIN_PHONE;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("ctl_process during LOGIN_PHONE write result to user");
							usr_exit(ptr);
							continue;
						}

						//如果登陆失败  关闭socket
						if( ptr->reply != 0 )		
						{
							close(ptr->fd);
							delete ptr;
							continue;
						}

					}
					else if( pkg_in->type == REQUEST_LIST )		//手机用户请求列表
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;
		
#ifdef DEBUG
						cout << ptr->usr_name << "\trequest list\t" << pkg_in->str.path << endl;
#endif

						//获取所请求目录的成员信息
						get_list(pkg_in,ptr);
						delete pkg_in;	
						
						//发送List给客户端
						for( int i = 0 ; i < ptr->items.size() ; i++ )
						{
							write(ptr->fd,ptr->items[i],sizeof(item_info_t));
							delete ptr->items[i];
						}
						
						ptr->items.clear();
					}
					else if( pkg_in->type == CREATE_D ) //用户创建目录
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;

						//创建目录
						ptr->reply = mkdir(pkg_in->str.path,0766);
						if( ptr->reply < 0 )
						{
							cerr << "ctl_process during CREATE_D " << pkg_in->str.path << " ";
							perror(NULL);
							delete pkg_in;
						}
						else		//若操作成功  则加入日志中
						{
#ifdef DEBUG
							cout << ptr->usr_name << "\tcreat dir\t"
								 <<	pkg_in->str.path << endl;
#endif
							ptr->history.push_back(pkg_in);
						}
						
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = CREATE_D;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("ctl_process during CREAT_D write to user");
							usr_exit(ptr); //用户异常退出
							continue;
						}

					}
					else if( pkg_in->type == CREATE_F ) //用户创建文件
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;

						//创建目录
						int new_file = open(pkg_in->str.path,O_CREAT|O_RDWR|O_EXCL,0766);
						if( new_file < 0 )
						{
							cerr << "ctl_process during CREATE_F " << pkg_in->str.path << " ";
							perror(NULL);
							ptr->reply = -1;
							delete pkg_in;
						}
						else
						{
							close(new_file);
							ptr->reply = 0;
#ifdef DEBUG
							cout << ptr->usr_name << "\tcreat file\t"
								 <<	pkg_in->str.path << endl;
#endif
							
							//若操作成功  则加入日志中
							ptr->history.push_back(pkg_in);
						}
						
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = CREATE_F;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("ctl_process during CREAT_F write to user");
							usr_exit(ptr); //用户异常退出
							continue;
						}
						
					}
					else if( pkg_in->type == DEL_F ) //用户删除文件
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;

						//删除文件
						ptr->reply = unlink(pkg_in->str.path);
						if( ptr->reply < 0 )
						{
							cerr << "ctl_process during DEL_F " << pkg_in->str.path << " ";
							perror(NULL);
							delete pkg_in;
						}
						else		//若操作成功  则加入日志中
						{
#ifdef DEBUG
							cout << ptr->usr_name << "\tremove file\t"
								 <<	pkg_in->str.path << endl;
#endif
							ptr->history.push_back(pkg_in);
						}
						
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = DEL_F;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("ctl_process during DEL_F write to user");
							usr_exit(ptr); //用户异常退出
							continue;
						}

					}
					else if( pkg_in->type == DEL_D ) // 用户删除目录
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;

						//删除文件
						ptr->reply = dfs_rmdir(pkg_in->str.path);
						if( ptr->reply < 0 )
						{
							cerr << "ctl_process during DEL_F " << pkg_in->str.path << " ";
							perror(NULL);
							delete pkg_in;
						}
						else	//若操作成功  则加入日志中
						{
#ifdef DEBUG
							cout << ptr->usr_name << "\tremove dir\t"
								 <<	pkg_in->str.path << endl;
#endif
							ptr->history.push_back(pkg_in);
						}
						
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = DEL_D;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("ctl_process during DEL_D write to user");
							usr_exit(ptr); //用户异常退出
						}

					}
					else if( pkg_in->type == MOVE )		//移动文件、文件夹
					{
						ctl_pkg_t *to = new ctl_pkg_t;
						
						//读取下一个包
						setblocking(ptr->fd);
						int res;
						
						if( (res = readn(ptr->fd, to, sizeof(ctl_pkg_t))) <= 0 )
						{
							if( res == 0 )
							{
								//对方关闭socket 用户下线
								usr_exit(ptr);
								delete pkg_in;
								delete to;
								continue;
							} 
							else
							{
								cerr << "ctl_process during MOVE read next pkg ";
								perror(NULL);
								usr_exit(ptr);
								delete pkg_in;
								delete to;
								continue;
							}
						}

						setnonblocking(ptr->fd);

						//重命名
						if( (ptr->reply = rename(pkg_in->str.path, to->str.path)) < 0 )
						{
							cerr << "ctl process during MOVE from " << pkg_in->str.path
									<< " to " << to->str.path << " ";
							perror(NULL);
							delete pkg_in;
							delete to;
						}
						//若操作成功  输出信息并且保存日志
						if(ptr->reply == 0)
						{
#ifdef	DEBUG
							cout << ptr->usr_name << "\trename\t\t" << pkg_in->str.path 
									<< " to " << to->str.path << endl;
#endif
							ptr->history.push_back(pkg_in);
							ptr->history.push_back(to);
						}

						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = MOVE;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("ctl_process during MOVE write to user");
							usr_exit(ptr); //用户异常退出
							continue;
						}

					}
					else if( pkg_in->type == UP_F)		//更新文件
					{
						//防止空串
						if( strlen(pkg_in->str.path) == 0 )
							continue;

#ifdef DEBUG
						cout << ptr->usr_name << "\tupdate file\t"
							 <<	pkg_in->str.path << endl;
#endif
						ptr->reply = 0;
						ptr->history.push_back(pkg_in);
						
						//写回结果
						if( write(ptr->fd,&(ptr->reply),sizeof(int8_t)) < 0 )
						{
							if( errno == EAGAIN || errno == EWOULDBLOCK )
							{
								ptr->state = DEL_D;
								events[i].events = EPOLLOUT | EPOLLET;
								epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&events[i]);
								continue;
							}
							perror("ctl_process during UP_F write to user");
							usr_exit(ptr); //用户异常退出
							continue;
						}

					}
					else if( pkg_in->type ==  EXIT)
					{
						usr_exit(ptr);
					}


				}
				else if( events[i].events & EPOLLOUT)
				{
					if( write(ptr->fd,&ptr->reply,sizeof(int8_t)) < 0 )
					{
						perror("ctl_process in EPOLLOUT  write reply to client");
						usr_exit(ptr);
						continue;
					}

					if( ptr->state == REG )
					{
						close(ptr->fd);
						delete(ptr);
						continue;
					}
					else if(ptr->state == LOGIN )
					{
						if( ptr->reply < 0 )	//如果登录失败
						{
							close(ptr->fd);
							delete(ptr);
							continue;
						} 
						else
						{
							//如果登录成功 发送版本号
							setblocking(ptr->fd);
							
							if(write(ptr->fd,&(ptr->version),sizeof(ptr->version)) < 0)
							{
								usr_exit(ptr);
								continue;
							}

							setnonblocking(ptr->fd);
						}
					}
					
					ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
					ev.data.ptr = ptr;
					if( epoll_ctl(epfd,EPOLL_CTL_MOD,ptr->fd,&ev) < 0 )
					{
						perror("epoll_ctl");
						usr_exit(ptr);
					}
				}
				else if( events[i].events & EPOLLERR)
				{
					//返回EPOLLERR事件说明
					//客户端心跳超时 
					//于是可以判定用户掉线
					usr_exit(ptr);
				}
				else if( events[i].events & EPOLLRDHUP )
				{
					//对方关闭socket
					usr_exit(ptr);
				}
			}
		}
		
	}

}

/* ---------------------------------------------------------------*/
/**
 * 函数简介:  接受控制信息的进程，控制信息包括新建、删除
 *            更新等
 *
 * 参数：n_child 要创建的worker数
 */
/* ---------------------------------------------------------------*/
void ctl_workers(int n_child)
{
	int listen_sock_ctl = tcp_listen(NULL,SRV_CTL_PORT,NULL);	
	if( listen_sock_ctl < 0 )
	{
		perror("tcp_listen");
		ctl_exit(-1);
	}

	int listen_sock_notify = tcp_listen(NULL,SRV_NOTIFY_PORT,NULL);
	if( listen_sock_notify < 0 )
	{
		perror("tcp_listen");
		master_exit(-1);
	}

	setnonblocking(listen_sock_ctl);
	setnonblocking(listen_sock_notify);
	
	//创建多个进程拷贝
	create_workers(n_child,1);

	ctl_process(listen_sock_ctl,listen_sock_notify);
}



/* ---------------------------------------------------------------*/
/**
 * 函数简介:  主函数，初始化各worker进程
 *
 * 参数: argc
 * 参数: argv[]
 *
 * 返回值:   
 */
/* ---------------------------------------------------------------*/
int main(int argc, char *argv[])
{
	//连接数据库
	if( db.init() < 0 )
	{
		exit(-1);
	}
	
	online_info = &online;

	if(online_info == NULL)
	{
		perror("mmap");
		exit(-1);
	}

	//分裂成两个进程
	pid_t pid = fork();
	if( pid < 0 )
	{
		perror("fork error");
		exit(0);
	}
	else if( pid == 0 )
	{
		//创建多个data进程
		//data进程用来处理用户的上传、下载、登录同步这3种请求
		data_workers(2);
	}
	else
	{
		//创建多个控制进程
		//控制进程处理不需要传输大数据
		ctl_workers(1);
	}
}

